
#include "runtime/network_poller.h"
#include "util/logger.h"

using namespace kawauso;
using namespace std;

#define PACKET_BUFFER_POOL_SIZE 8192
#define PACKET_BUFFER_POOL_CACHE_SIZE 32

NetworkPoller::NetworkPoller(NetworkOptions &options)
    : packetBufferPools(rte_lcore_count())
    , options(options)
{
    initPacketBufferPools();
    initEthernetDevices();
}

void NetworkPoller::initPacketBufferPools() {
    uint16_t coreId;
    RTE_LCORE_FOREACH(coreId) {
        string poolName = fmt::format("packet_buffer_pool_{}", coreId);

        packetBufferPools[coreId] = rte_pktmbuf_pool_create(poolName.c_str(),
                        PACKET_BUFFER_POOL_SIZE, PACKET_BUFFER_POOL_CACHE_SIZE, 0, RTE_MBUF_DEFAULT_BUF_SIZE,
                        rte_lcore_to_socket_id(coreId));

        if (packetBufferPools[coreId] == nullptr) {
            throw std::runtime_error("Failed to create packet buffer pool");
        }
    }
}

void NetworkPoller::initEthernetDevices() {

    rte_eth_dev_info deviceInfo{};
    uint16_t portId;

    RTE_ETH_FOREACH_DEV(portId) {

        int res = rte_eth_dev_info_get(portId, &deviceInfo);
        if (res < 0) {
            throw std::runtime_error(fmt::format(
                    "Failed to get device info (port {})", portId));
        }

        EthernetDevice &device = devices.emplace_back(portId, rte_lcore_count(), deviceInfo);
        device.init(packetBufferPools);
    }
}

bool NetworkPoller::poll() {

    uint16_t coreId = rte_lcore_id();
    uint16_t totalRxPollCount = 0;
    uint16_t emptyRxPollCount = 0;

    for (auto &device : devices) {
        if (coreId < device.getQueueCount()) {
            ReceiveQueue &rxQueue = device.getRxQueue(coreId);
            vector<rte_mbuf *> packets = rxQueue.receivePackets();

            if (unlikely(packets.empty())) {
                rxQueue.emptyRxPollCount ++;
                if (rxQueue.emptyRxPollCount >= options.emptyRxPollThreshold) {
                    emptyRxPollCount ++;
                }
            } else {
                for (auto *packet : packets) {
                    // TODO 根据ethernet type code找到对应的网络层协议栈进行处理
                    info("core id: {}, packet len: {}", rte_lcore_id(), packet->data_len);
                    rte_pktmbuf_free(packet);
                }
                rxQueue.emptyRxPollCount = 0;
            }

            totalRxPollCount ++;
        }
    }

    return emptyRxPollCount < totalRxPollCount;
}
